package com.fanshuai.algorithms.divide;

import java.io.File;
import java.io.RandomAccessFile;
import java.util.*;

/**
 * 大文件，寻找相同的url。
 * 有2个文件，A文件和B文件都有50亿行，每行一个url，url的长度是64个字节。
 * 假设机器内存有4G，设计程序寻找相同的url
 *
 * 使用分治法，寻找不同的文件区间的相同url，最后合并
 * 递归出口：估算机器内存能够存放的最大url数量。50亿行大约320G，使用二分法，需要拆分8次，拆分的最小单元数量为：320G/128
 *
 * 最小的文件单元比较：
 * A文件，区间为[startA, endA]，B文件区间为[startB, endB]。
 * 1 可以建立A文件区间和b文件区间的hash表，然后比较
 * 2 可以对A文件区间建立布隆过滤器，然后扫描B区间，找到A文件区间可能包含的url，结果集为list1（布隆过滤器有误差）。
 * 然后A文件区间建立hash表，过滤不包含的部分，得到结果集list2
 */
public class BigFileSameUrl {

    private static RandomAccessFile getRandomFile(String file) {
        try {
            return new RandomAccessFile(new File(file), "r");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private static String readLine(RandomAccessFile file, long line)  {
        long start = line * (64 + System.lineSeparator().length());

        try {
            file.seek(start);
            return file.readLine();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private static long lines = 5000000000L;
    private static long minPartition = 64L * lines / 128L;  //分隔的文件最小区间行数
    public static List<String> findSameUrl(String fileA, String fileB) {
        RandomAccessFile randomAccessFileA = getRandomFile(fileA);
        RandomAccessFile randomAccessFileB= getRandomFile(fileB);

        if (null == fileA || null == fileB) {
            return new ArrayList<>();
        }

        return findSameUrl(randomAccessFileA, randomAccessFileB, 0, lines - 1, 0, lines - 1);
    }

    /**
     * 最小区间  叶节点操作
     * @param fileA  文件A
     * @param fileB  文件B
     * @param startA 文件A开始行数
     * @param endA   文件A结束行数
     * @param startB 文件B开始行数
     * @param endB   文件B结束行数
     * @return
     */
    private static List<String> findSameUrl(RandomAccessFile fileA, RandomAccessFile fileB,
                                            long startA, long endA, long startB, long endB) {

        if (endA - startA <= minPartition && endB - startB <= minPartition) {
            return findSameUrl0(fileA, fileB, startA, endA, startB, endB);
        }

        List<String> sameUrls = new ArrayList<>();

        long midA = (startA + endA) / 2, midB = (startB + endB) / 2;
        sameUrls.addAll(findSameUrl(fileA, fileB, startA, midA, startB, midB));
        sameUrls.addAll(findSameUrl(fileA, fileB, startA, midA, midB + 1, endB));
        sameUrls.addAll(findSameUrl(fileA, fileB, midA + 1, endA, startB, midB));
        sameUrls.addAll(findSameUrl(fileA, fileB, midA + 1, endA, midB + 1, endB));

        return sameUrls;
    }

    /**
     * 最小区间  叶节点操作
     * @param fileA  文件A
     * @param fileB  文件B
     * @param startA 文件A开始行数
     * @param endA   文件A结束行数
     * @param startB 文件B开始行数
     * @param endB   文件B结束行数
     * @return
     */
    private static List<String> findSameUrl0(RandomAccessFile fileA, RandomAccessFile fileB,
                                            long startA, long endA, long startB, long endB) {

        Set<String> linesA = new HashSet<>();
        for (long i = startA; i <= endA; i++) {
            String line = readLine(fileA, i);
            if (null != line) {
                linesA.add(line);
            }
        }
        List<String> sameUrl = new ArrayList<>();
        for (long i = startB; i <= endB; i++) {
            String line = readLine(fileB, i);
            if (null != line && linesA.contains(line)) {
                sameUrl.add(line);
            }
        }

        return sameUrl;
    }
}
